package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.util.{DateUtil, KafkaUtil, NoticeMsg, SparkUtil}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
 * Created by Lenovo on 2018/8/12.
 * 18.多发服药后低糖警报：连续两个点血糖<3.0mmol/L,5天内出现3次&服药后1小时内：low_glycemia_drug_freq：每次用药后：触发及时发送，每小时提示不超过1条
 * 24.服药后低糖：连续两个点血糖<2.8mmol/L&服药后1小时内：low_glycemia_drug：每次用药后：触发及时发送
 * 32.服药后降糖过速：血糖下降变化超过6.0mmol/L/h&用药后1小时内：slope_glycemia_drug：每次用药：触发及时发送
 * 总计3个
 * {
 * "personId": 1000000001007744,
 * "actualDate": 1533727800000,
 * "drugList": [
 * {
 * "name": "阿卡波糖片",
 * "dosage": "3粒/天"
 * },
 * {
 * "name": "亚莫利格列美脲片",
 * "dosage": "3粒/天"
 * },
 * {
 * "name": "23",
 * "dosage": ">15U/天"
 * },
 * {
 * "name": "诺和灵N",
 * "dosage": ">15U/天"
 * }
 * ],
 * "createTime": 1534233432151
 * }
 */
object BloodSugarDrugStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     */
    println(args.mkString("\t"))
    val Array(brokers, topicList, otherTopicList, mqHost, mqPort, mqUsername, mqPassword,
    mqVirtualHost, exchangeName, mqRouteKey, checkpointDir, duation, rddDuration, appName
    ) = args


    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord], classOf[DrugRecord]))
    val sc = SparkContext.getOrCreate(conf)
    //    val checkpointDir = "/data/streaming/checkpoint/blood_sugar_drug_stream/"

    val updateSugarFunc = (values: Seq[ArrayBuffer[BloodSugarRecord]],
                           state: Option[ArrayBuffer[BloodSugarRecord]]) => {
      //保存状态
      val HourTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 12 * 60 * 60 * 1000
      val currentCount = ArrayBuffer[BloodSugarRecord]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(ArrayBuffer[BloodSugarRecord]())
      Some(currentCount ++ previousCount.filter(_.timestamp > HourTimeStamp))
    }
    val updateDrugFunc = (values: Seq[ArrayBuffer[DrugRecord]],
                          state: Option[ArrayBuffer[DrugRecord]]) => {
      //保存状态
      val HourTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 12 * 60 * 60 * 1000
      val currentCount = ArrayBuffer[DrugRecord]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(ArrayBuffer[DrugRecord]())
      Some(currentCount ++ previousCount.filter(_.actualDate > HourTimeStamp))
    }

    val updateDrugTimeStateFunc =
      (values: Seq[mutable.HashMap[Long, Int]],
       state: Option[mutable.HashMap[Long, Int]]) => {
        //保存状态
        val fiveDaysAgoTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 5 * 24 * 60 * 60 * 1000
        val currentState = mutable.HashMap[Long, Int]()
        val previousState = state.getOrElse(mutable.HashMap[Long, Int]())
        values.foreach {
          case map =>
            map.foreach { case (drugTime, flag) =>
              if ((previousState.contains(drugTime) && previousState(drugTime) == 1)
                || (currentState.contains(drugTime) && currentState(drugTime) == 1)
              ) {
                currentState += ((drugTime, 1))
              } else {
                currentState += ((drugTime, flag))
              }
            }
        }
        val tempOldState = previousState.filter { case (drugTime, flag) =>
          !currentState.contains(drugTime) && drugTime > fiveDaysAgoTimeStamp
        }
        Some(currentState ++= tempOldState)
      }

    def creatingFunc(): StreamingContext = {
      val ssc = new StreamingContext(sc, Seconds(duation.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn = new MQObject(mqHost, mqPort.toInt, mqUsername, mqPassword, mqVirtualHost)

      val sugarTopics = topicList.split(",").toSet
      val drugTopics = otherTopicList.split(",").toSet
      val sugarMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc,kafkaParams,sugarTopics)
      val drugMessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc,kafkaParams, drugTopics)

      val sugarBatch: DStream[(String, ArrayBuffer[BloodSugarRecord])] = sugarMessages.filter(_._2.nonEmpty)
        .flatMap { msg =>
          try {
            println("___________ msg :" + msg)
            val gson = new Gson()
            val voList =
              gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
            voList.map { vo =>
              (vo.personId.toString, ArrayBuffer[BloodSugarRecord](vo))
            }.toList
          } catch {
            case e: Exception =>
              println("____________" + e.getMessage)
              List(("", ArrayBuffer[BloodSugarRecord]()))
          }
        }
        .filter(_._1.nonEmpty)
        .updateStateByKey(updateSugarFunc)

      val drugBatch: DStream[(String, ArrayBuffer[DrugRecord])] = drugMessages.filter(_._2.nonEmpty)
        .map { msg =>
          try {
            println("___________ msg :" + msg)
            val gson = new Gson()
            val vo = gson.fromJson(msg._2, classOf[DrugRecord])
            (vo.personId.toString, ArrayBuffer[DrugRecord](vo))
          } catch {
            case e: Exception =>
              println("____________" + e.getMessage)
              ("", ArrayBuffer[DrugRecord]())
          }
        }
        .filter(_._1.nonEmpty)
        .updateStateByKey(updateDrugFunc)

      sugarBatch.checkpoint(Duration(rddDuration.toInt * 1000))
      drugBatch.checkpoint(Duration(rddDuration.toInt * 1000))

      val sugarAndDrug = drugBatch.leftOuterJoin(sugarBatch)
      val sortRdd = sugarAndDrug.filter { case (pid, (drugList, sugarList)) =>
        sugarList.isDefined && drugList.nonEmpty && sugarList.nonEmpty
      }.map {
        case (pid, (drugList, sugarList)) =>
          (
            pid,
            (
              drugList.sortBy(_.actualDate),
              sugarList.get.sortBy(_.timestamp)
              )
            )
      }

      //@todo （多发服药后低糖记录）
      val drugTimeDoneFlag = sortRdd
        .map {
          case (pid, (drugList, sugarList)) =>
            val result = mutable.HashMap[Long, Int]()
            drugList.foreach { drugVo =>
              result += ((drugVo.actualDate, 0))
              val afterDrugOneHour = drugVo.actualDate + 60 * 60 * 1000
              val afterRangeTimeSugar = sugarList.filter(
                sugarVo =>
                  sugarVo.timestamp >= drugVo.actualDate && sugarVo.timestamp < afterDrugOneHour
              )
              if (afterRangeTimeSugar.size >= 2) {
                var preVo = afterRangeTimeSugar.head
                for (i <- 1 to afterRangeTimeSugar.size - 1) {
                  val nowVo = afterRangeTimeSugar(i)
                  if (nowVo.timestamp - preVo.timestamp < 15 * 60 * 1000 &&
                    preVo.value < 3 && nowVo.value < 3) {
                    result += ((drugVo.actualDate, 1))
                  }
                  preVo = nowVo
                }
              }
            }
            (pid, result)
        }
        .updateStateByKey(updateDrugTimeStateFunc)

      drugTimeDoneFlag.checkpoint(Duration(rddDuration.toInt * 1000))

      //@todo 18.多发服药后低糖警报：连续两个点血糖<3.0mmol/L,5天内出现3次&服药后1小时内：low_glycemia_drug_freq：每次用药后：触发及时发送，每小时提示不超过1条
      drugTimeDoneFlag.foreachRDD { mess =>
        if (!mess.isEmpty()) {
          mess.foreachPartition { lines =>
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, drugListState) =>
                println("____ pid :" + pid + "____ drugListState :" + drugListState.toString())
                if (drugListState.values.sum >= 3) {
                  val msg = new WarningMsg(
                    "glycemia",
                    pid.toLong,
                    NoticeMsg.low_glycemia_drug_freq._1,
                    NoticeMsg.low_glycemia_drug_freq._2,
                    NoticeMsg.low_glycemia_drug_freq._3,
                    drugListState.keys.max.toString,
                    NoticeMsg.low_glycemia_drug_freq._5,
                    NoticeMsg.low_glycemia_drug_freq._4,
                    NoticeMsg.low_glycemia_drug_freq._6
                  )
                  mqMsg += gson.toJson(msg)
                }
            }

            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }
          }
        }
      }

      sortRdd.foreachRDD(mess =>
        if (!mess.isEmpty()) {
          mess.foreachPartition(lines => {
            val mqMsg = ArrayBuffer[String]()
            val gson = new Gson()
            lines.foreach {
              case (pid, (drugList, sugarList)) => {
                println("____ pid :" + pid + "____ drugList:" + drugList.map(gson.toJson(_)).toString()
                  + "____ sugarList" + sugarList.map(gson.toJson(_)).toString()
                )

                drugList.foreach { drugVo =>
                  //@todo 24.服药后低糖：连续两个点血糖<2.8mmol/L&服药后1小时内：low_glycemia_drug：每次用药后：触发及时发送
                  val afterDrugOneHour = drugVo.actualDate + 60 * 60 * 1000
                  val rangeTimeSugar = sugarList.filter(
                    sugarVo =>
                      sugarVo.timestamp >= drugVo.actualDate && sugarVo.timestamp <= afterDrugOneHour
                  )
                  if (rangeTimeSugar.size >= 2) {
                    var preVo = rangeTimeSugar.head
                    for (i <- 1 to rangeTimeSugar.size - 1) {
                      val nowVo = rangeTimeSugar(i)
                      if (nowVo.timestamp - preVo.timestamp < 15 * 60 * 1000
                        && preVo.value < 2.8 && nowVo.value < 2.8) {
                        val msg = new WarningMsg(
                          "glycemia",
                          pid.toLong,
                          NoticeMsg.low_glycemia_drug._1,
                          NoticeMsg.low_glycemia_drug._2,
                          NoticeMsg.low_glycemia_drug._3,
                          drugVo.actualDate.toString,
                          NoticeMsg.low_glycemia_drug._5,
                          NoticeMsg.low_glycemia_drug._4,
                          NoticeMsg.low_glycemia_drug._6
                        )
                        mqMsg += gson.toJson(msg)
                      }
                      preVo = nowVo
                    }
                  }

                  //@todo 32.服药后降糖过速：血糖下降变化超过6.0mmol/L/h&用药后1小时内：slope_glycemia_drug：每次用药：触发及时发送
                  val inOneHourRangeTimeSugar = sugarList.filter(
                    sugarVo =>
                      sugarVo.timestamp <= afterDrugOneHour && sugarVo.timestamp > drugVo.actualDate
                  )
                  inOneHourRangeTimeSugar.foreach {
                    vo =>
                      val beforeThisOneHour = vo.timestamp - 60 * 60 * 1000
                      val beforeThisOneListMaxSugar = sugarList
                        .filter(
                          sugarVo =>
                            sugarVo.timestamp < vo.timestamp && sugarVo.timestamp >= beforeThisOneHour
                        )
                      if (beforeThisOneListMaxSugar.nonEmpty) {
                        if (beforeThisOneListMaxSugar.maxBy(_.value).value - vo.value > 6) {
                          val msg = new WarningMsg(
                            "glycemia",
                            pid.toLong,
                            NoticeMsg.slope_glycemia_drug._1,
                            NoticeMsg.slope_glycemia_drug._2,
                            NoticeMsg.slope_glycemia_drug._3,
                            drugVo.actualDate.toString,
                            NoticeMsg.slope_glycemia_drug._5,
                            NoticeMsg.slope_glycemia_drug._4,
                            NoticeMsg.slope_glycemia_drug._6
                          )
                          mqMsg += gson.toJson(msg)
                        }
                      }
                  }
                }
              }
            }
            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }

          })
        })
      ssc
    }

    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }
}